package xpulsar

import (
	"context"
	"crypto/rand"
	"encoding/json"
	"errors"
	"fmt"
	"gitee.com/go-mid/infra/xmq/mqconf"
	"io/ioutil"
	"math/big"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"time"

	"gitee.com/go-mid/infra/xtime"

	"gitee.com/go-mid/infra/xtrace"

	"github.com/apache/pulsar-client-go/pulsar"
)

var (
	ErrBrokerNotNull           = errors.New("broker cannot be null")
	ErrConfigNotNull           = errors.New("config cannot be nil")
	ErrTopicInfoNotNull        = errors.New("topic info cannot be null")
	ErrSubscriptionNameNotNull = errors.New("subscription name cannot be null")
	ErrSubNameNotChinese       = errors.New("subscription name cannot be chinese")
)

type Consumer struct {
	pulsar.Consumer
	client           pulsar.Client
	config           *ConsumerConfig
	subscriptionName string
	url              string
	topic            string
}

func NewConsumer(ctx context.Context, config *ConsumerConfig) (*Consumer, error) {
	fun := "NewConsumer -->"

	if err := validateConsumerConfig(config); err != nil {
		return nil, err
	}

	client, err := pulsar.NewClient(pulsar.ClientOptions{
		URL:               config.Broker,
		ConnectionTimeout: config.ConnectionTimeout,
		OperationTimeout:  config.OperationTimeout,
	})
	if err != nil {
		return nil, fmt.Errorf("%s new client, err: %v", fun, err)
	}

	consumer, err := client.Subscribe(*config.ConsumerOptions)
	if err != nil {
		client.Close()
		return nil, fmt.Errorf("%s subscribe, err: %v", fun, err)
	}
	return &Consumer{
		Consumer:         consumer,
		client:           client,
		subscriptionName: config.SubscriptionName,
		url:              config.Broker,
		topic:            config.Topic,
		config:           config,
	}, err
}

func (c *Consumer) setSpanTags(span xtrace.Span) {
	span.SetTag(xtrace.TagComponent, mqconf.TraceComponent)
	span.SetTag(xtrace.TagPalfishMessageBusType, mqconf.TraceMessageBusTypePulsar)
	span.SetTag(xtrace.TagSpanKind, xtrace.SpanKindConsumer)
	span.SetTag(xtrace.TagMessageBusDestination, c.topic)
	span.SetTag(xtrace.TagMessagingSystem, xtrace.MessagingSystemPulsar)
	span.SetTag(xtrace.TagMessagingDestinationKind, xtrace.MessagingDestinationKindTopic)
	span.SetTag(xtrace.TagMessagingDestination, c.topic)
	span.SetTag(xtrace.TagMessagingPulsarConsumerGroup, c.subscriptionName)
	span.SetTag(xtrace.TagNetPeerIP, c.url)
}

func (c *Consumer) ReadMsg(ctx context.Context, v interface{}) error {
	span := xtrace.SpanFromContext(ctx)
	if span != nil {
		c.setSpanTags(span)
	}

	st := xtime.NewTimeStat()
	msg, err := c.Receive(ctx)
	mqconf.StatReqDuration(ctx, c.topic, "Consumer.ReadMsg", mqconf.TraceMessageBusTypePulsar, st.Millisecond())

	if err != nil {
		return err
	}
	defer c.Ack(msg)

	err = json.Unmarshal(msg.Payload(), v)
	if err != nil {
		return err
	}

	return err
}

func (c *Consumer) FetchMsg(ctx context.Context, v interface{}) (mqconf.AckHandler, error) {
	span := xtrace.SpanFromContext(ctx)
	if span != nil {
		c.setSpanTags(span)
	}

	st := xtime.NewTimeStat()
	msg, err := c.Receive(ctx)
	mqconf.StatReqDuration(ctx, c.topic, "Consumer.FetchMsg", mqconf.TraceMessageBusTypePulsar, st.Millisecond())
	if err != nil {
		return nil, err
	}
	err = json.Unmarshal(msg.Payload(), v)
	if err != nil {
		return nil, err
	}

	ackHandler := newPulsarHandler(msg, c.Consumer)
	return ackHandler, nil
}

func (c *Consumer) Close(ctx context.Context) error {
	c.Consumer.Close()
	c.client.Close()

	return nil
}

func (c *Consumer) SetOffsetAt(ctx context.Context, t time.Time) error {
	fun := "SetOffsetAt -->"
	if c.config.BrokerAdmin == "" {
		return fmt.Errorf("%s invalid BrokerAdmin", fun)
	}
	topics := strings.Split(c.topic, "://")
	if len(topics) != 2 {
		return fmt.Errorf("%s invalid Topic", fun)
	}
	res, err := http.Post(fmt.Sprintf("%s/admin/v2/persistent/%s/subscription/%s/resetcursor/%d",
		c.config.BrokerAdmin,
		topics[1],
		c.Subscription(),
		t.UnixNano()/1e6), "application/json", nil)
	if err != nil {
		return err
	}
	defer res.Body.Close()
	body, _ := ioutil.ReadAll(res.Body)
	if string(body) != "" {
		return fmt.Errorf("%s error:%s", fun, string(body))
	}
	return nil
}

func (c *Consumer) SetOffset(ctx context.Context, offset interface{}) error {
	switch offset.(type) {
	case pulsar.MessageID:
		return c.Consumer.Seek(offset.(pulsar.MessageID))
	}
	return nil
}

func getName() string {
	progname := filepath.Base(os.Args[0])
	hostname, _ := os.Hostname()
	num, _ := rand.Int(rand.Reader, big.NewInt(10000))
	return fmt.Sprintf("%s@%s (gitee.com/go-mid/infra/xmq) %d", progname, hostname, num)
}

func validateConsumerConfig(config *ConsumerConfig) error {
	if config == nil {
		return ErrConfigNotNull
	}
	if config.Broker == "" {
		return ErrBrokerNotNull
	}
	if config.Topic == "" && len(config.Topics) == 0 {
		return ErrTopicInfoNotNull
	}
	if config.SubscriptionName == "" {
		return ErrSubscriptionNameNotNull
	}
	return nil
}

type pulsarHandler struct {
	msg      pulsar.Message
	consumer pulsar.Consumer
}

func (p *pulsarHandler) Ack(ctx context.Context) error {
	p.consumer.Ack(p.msg)
	return nil
}

func newPulsarHandler(msg pulsar.Message, consumer pulsar.Consumer) *pulsarHandler {
	return &pulsarHandler{
		msg:      msg,
		consumer: consumer,
	}
}
